package nmp.lib;

import java.io.IOException;
import java.io.InputStream;
import java.io.InterruptedIOException;
import java.io.OutputStream;
import java.net.DatagramPacket;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

import android.os.Bundle;
import android.os.Handler;
import android.os.Message;
import android.util.Log;

public class CNmpTcp 
{
	public static final boolean CLIENT = false;
	public static final boolean SERVER = true;

	protected static final String BUNDLE_DATA_STRING = "bundle_data_string";
	protected static final String BUNDLE_DATA_ADDRESS = "bundle_data_address";
	protected static final String BUNDLE_DATA_PORT = "bundle_data_port";
	
	protected static final String NMP_SYNC_WORD = "nmp";
	
	protected static final int MSG_NET_STATUS_DISCONNECT = 0;
	protected static final int MSG_NET_STATUS_CONNECT = 1;
	protected static final int MSG_GET_MESSAGE = 2;
	protected static final int MSG_NET_MESSAGE_TIMEOUT = 3;
	protected static final int MSG_NET_CONNECT_TIMEOUT = 4;
	
	protected static final int IN_BUFFER_SIZE = 1024;
	
	private AtomicBoolean m_thread_run = new AtomicBoolean(false);
	
	private AtomicInteger m_status = new AtomicInteger(MSG_NET_STATUS_DISCONNECT);
	
	private Socket m_socket = null; //for tcp client
	private ServerSocket m_server = null; // for tcp server 
    private Thread m_read_thread = null;
    private Thread m_write_thread = null;
    private Handler m_handler = null;
    private int m_timeout = 0;
    private int m_connect_timeout = 0;
    private boolean m_f_timeout_msg = false;
    
    private boolean m_isReady;    
    private boolean m_isServer;
    private int m_bind_port;
    private String m_bind_address;
    private int m_remote_port;
    private String m_remote_address;
    private BlockingQueue<String> m_write_queue;
    ByteBuffer m_syncword = null;
    ByteBuffer m_synclen = null;
    OutputStream m_outputStream = null;
    InputStream m_inputStream = null;
    
    public CNmpTcp(boolean isServer,String address,int port)
    {
    	m_isServer = isServer;
    	m_isReady = false;
    	m_f_timeout_msg = false;
    	if(m_isServer)
    	{	
    		m_bind_port = port;
    		m_bind_address = address;
    	}
    	else
    	{
    		m_remote_port = port;
    		m_remote_address = address;
    	}
    	m_write_queue = new ArrayBlockingQueue<String>(20,true);
    	m_syncword = ByteBuffer.allocate(4);    	
    	m_syncword.put((byte) 'n');
    	m_syncword.put((byte) 'm');
    	m_syncword.put((byte) 'p');
    	m_syncword.put((byte) 0);
    	m_syncword.rewind();
    	m_synclen = ByteBuffer.allocate(4);
    }

    public void setTimeout(int timeout,int connect_timeout,
    		boolean f_timeout_msg)
    {
    	m_timeout = timeout;   
    	m_connect_timeout = connect_timeout;
    	m_f_timeout_msg = f_timeout_msg;
    }
    
    public void start()
    {
        m_read_thread = new Thread()
        { 
            @Override
            public void run()
            { 
            	if(m_isServer) {
					try {
						m_server = new ServerSocket(m_bind_port);
					} catch (IOException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
						m_thread_run.set(false);
					}
            	}
    			
            	while(m_thread_run.get()) {
            		
					try {
						
						if(m_isServer) {
							Log.d(this.toString(),"TCP server wait to accetp a connection...");
							m_socket = m_server.accept();							
						}  else {
							m_socket = new Socket();							
							if(m_remote_address == null)
								m_remote_address = m_socket.getLocalAddress().getHostAddress();
							SocketAddress remoteAddress = new InetSocketAddress(m_remote_address, m_remote_port);
							Log.e(this.toString(),"TCP client try to connect to :" + m_remote_address + ":" + m_remote_port);							
							m_socket.connect(remoteAddress,m_connect_timeout);
						}
						
    					m_remote_address = m_socket.getInetAddress().getHostAddress();
    					m_remote_port = m_socket.getPort();	
    					    				    						
	        			m_bind_address = m_socket.getLocalAddress().getHostAddress();
	        			m_bind_port = m_socket.getLocalPort();
	        			
	        			Log.e(this.toString(),"TCP connect to :" + m_remote_address + ":" + m_remote_port);
	        			Log.e(this.toString(),"TCP bind to :" + m_bind_address + ":" + m_bind_port);
	        			
	    				m_socket.setReuseAddress(true);  
	    				m_socket.setSoTimeout(m_timeout);
	    			
	    				m_inputStream = m_socket.getInputStream();
						m_outputStream = m_socket.getOutputStream();	    				

    					NoticeStatus(MSG_NET_STATUS_CONNECT);
    					
					} catch (IOException e) {
						// TODO Auto-generated catch block
    					Log.e(this.toString(),"Connection failed..");								
						e.printStackTrace();
						if(m_isServer) {
							NoticeStatus(MSG_NET_CONNECT_TIMEOUT);
							continue;
						} else {														
							NoticeStatus(MSG_NET_CONNECT_TIMEOUT);
							m_thread_run.set(false);
						}
					}

	                try {  
	                	
	                	ByteBuffer tmp = ByteBuffer.allocate(4);
	                	int readlen = 0;
	                	int datalength;
	                	
	                	while(m_thread_run.get()) {
	                		
	                		tmp.rewind();
	                		readlen = m_inputStream.read(tmp.array());
	                			                		
	                		if(readlen != 4) { 
	                			Log.e(this.toString(),"syncword length should be 4, but get data length:"+readlen);
	                			break;
	                	    }
		                			                	
		                	tmp.rewind();
		                	m_syncword.rewind();
		                	if(!tmp.equals(m_syncword)) {
		                		
		                		Log.e(this.toString(),"syncword error !!!!");
		                		continue;
		                	}
		                	
		                	tmp.rewind();
		                	readlen =  m_inputStream.read(tmp.array());
		                	
		                	datalength = tmp.getInt();
		                	
		                	ByteBuffer msg_buffer = ByteBuffer.allocate(datalength);
		                	
		                	readlen =  m_inputStream.read(msg_buffer.array());
		                	
		                	if(readlen != datalength) {
		                		Log.e(this.toString(),"read size:"+readlen+"!="+ "data length:"+datalength);
		                		continue;
		                	}
		                	
		                	String packetString = new String(msg_buffer.array(),0,msg_buffer.array().length-1);
		        
		    	            Message msg = new Message();
		    	            Bundle data = new Bundle();	 
		    	            
		    	            data.putString(BUNDLE_DATA_STRING,packetString);
		    	            data.putString(BUNDLE_DATA_ADDRESS,m_remote_address);
		    	            data.putInt(BUNDLE_DATA_PORT,m_remote_port);
		    	            msg.what = MSG_GET_MESSAGE;
		    	            msg.setData(data);
		    	            m_handler.sendMessage(msg);
	                	}
	                } catch (InterruptedIOException e) {
    	        		Log.e(this.toString(),"client get message timeout!!!");
    	        		if(m_f_timeout_msg) {
    	        			NoticeStatus(MSG_NET_MESSAGE_TIMEOUT);
    	        		}    	        		
	    	        } catch (Exception e) {
	                	Log.e(this.toString(),"finally!!!");
	                    e.printStackTrace();
	                } finally {
	                	
	                }
            		
            		NoticeStatus(MSG_NET_STATUS_DISCONNECT);
            		
            		if(!m_isServer) {
            			m_thread_run.set(false);
            		}
            	}            
            }
        }; 
        
        m_handler = new Handler() {  
           Bundle data;
           String data_str;
           String data_addr;
           int data_port;
           
    	   @Override  
    	   public void handleMessage(Message msg) {   
    	      switch (msg.what)
    	      {
    	      	case MSG_NET_STATUS_DISCONNECT:
    	      	case MSG_NET_STATUS_CONNECT:
    	      	case MSG_NET_CONNECT_TIMEOUT:
    	      	case MSG_NET_MESSAGE_TIMEOUT:	
    	      		data = msg.getData();
                	data_addr = data.getString(BUNDLE_DATA_ADDRESS);
                	data_port = data.getInt(BUNDLE_DATA_PORT);        	      		
    	      		UpdateStatus(msg.what,data_addr,data_port);
    	      		break;
    	      		
                case MSG_GET_MESSAGE:
                	data = msg.getData();
                	data_str = data.getString(BUNDLE_DATA_STRING);
                	data_addr = data.getString(BUNDLE_DATA_ADDRESS);
                	data_port = data.getInt(BUNDLE_DATA_PORT);                	
                	GetString(data_str,data_addr,data_port);
                    break;
	
                default:
                	break;
              }    		   
    	      
    	      super.handleMessage(msg);  
    	   }  
        };  
        
        
        m_write_thread =  new Thread() {
        	String data;
            public void run(){
            	while(true){
            		
            		if(!m_thread_run.get()){
            			break;
            		}
	            	try {
	            		data = m_write_queue.take();
					} catch (InterruptedException e1) {
						// TODO Auto-generated catch block
						e1.printStackTrace();
					}
	            	
            		if(!m_thread_run.get()){
            			break;
            		}
 
            		try {
                		m_outputStream.write(m_syncword.array(),0,4);
                		m_synclen.rewind();
                		m_synclen.putInt( data.length()+1 );                   		
						m_outputStream.write(m_synclen.array());
						m_outputStream.write(data.getBytes());
						m_outputStream.write((int)0); //end of string
						m_outputStream.flush();	
					} catch (IOException e) {
						// TODO Auto-generated catch block
						//UpdateStatus(MSG_NET_MESSAGE_TIMEOUT,null,0);
						e.printStackTrace();
					}
            	}
            }
		};
		
		m_thread_run.set(true);
        m_write_thread.start();
        m_read_thread.start();
        m_isReady = true;

    }
    
	public void SendString(String str)
	{		
		if(!m_isReady)
			return;
        try {
        	Log.e(this.toString(),"Send String1:" + str);
			m_write_queue.put(str);
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block			
			e.printStackTrace();
		}
	}
	
	public void GetString(String str, String from, int port)
	{
		Log.e(this.toString(),"Get string:" + str + "from: " + from + ":" + port);
	}
	
	public void UpdateStatus(int status, String address, int port) {
		switch(status) {
			case MSG_NET_STATUS_CONNECT:
				Log.e(this.toString(),"Connected:" + "from: " + address + ":" + port); 
				break;
			case MSG_NET_STATUS_DISCONNECT:
				Log.e(this.toString(),"Disconnected:" + "from: " + address + ":" + port); 
				break;
			case MSG_NET_MESSAGE_TIMEOUT:
				Log.e(this.toString(),"Message timeout:" + "from: " + address + ":" + port); 
				break;
			case MSG_NET_CONNECT_TIMEOUT:
				Log.e(this.toString(),"Connect timeout:" + "from: " + address + ":" + port); 
				break;							
			default:
				break;
		}		  
	}	
	
	private void NoticeStatus(int status)
	{
		Log.e(this.toString(),"status:" + status + "==" + m_status.get() );
		if(status == m_status.get())
			return;
        Message msg = new Message();   
        Bundle data = new Bundle();	 
        m_status.set(status);
        data.putString(BUNDLE_DATA_ADDRESS,m_remote_address);
        data.putInt(BUNDLE_DATA_PORT,m_remote_port);
        msg.what = status;
        msg.setData(data);
        m_handler.sendMessage(msg); 
	}
	
	public int GetStatus()
	{
        return m_status.get(); 
	}
	
	public void close()
	{						
		m_thread_run.set(false);
		
		try {
			if(m_socket != null)
				m_socket.close();			
			if(m_server != null)
				m_server.close();			
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
		try {
			m_read_thread.interrupt();	
			m_read_thread.join(3000);
			m_write_queue.put("");			
			m_write_thread.interrupt();	
			m_write_thread.join(3000);				
		} catch (InterruptedException e) {
			// TODO Auto-generated catch bloc k
			e.printStackTrace();
		}		
	}
 
    static String stringFromPacket(DatagramPacket packet) 
    {
        return new String(packet.getData(), 0, packet.getLength());        
    }    
}
